package org.lgq.parser;

import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.dataflow.props.DataflowJobProperties;
import org.apache.shardingsphere.elasticjob.http.props.HttpJobProperties;
import org.apache.shardingsphere.elasticjob.lite.api.bootstrap.impl.ScheduleJobBootstrap;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.apache.shardingsphere.elasticjob.script.props.ScriptJobProperties;

import org.lgq.annotation.ElasticJobConfig;
import org.lgq.constant.ElasticJobType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;

import java.util.Map;

/**
 * 解析es-job配置。要考虑在什么时机进行解析。在所有的spring bean都实例化完成之后
 *
 * @author DevLGQ
 * @version 1.0
 */
public class ElasticJobCfgParser implements ApplicationListener<ApplicationReadyEvent> {

    private static final Logger log = LoggerFactory.getLogger(ElasticJobCfgParser.class);
    public static final String PROTOTYPE = "prototype";
    private final CoordinatorRegistryCenter zookeeperRegistryCenter;

    public ElasticJobCfgParser(CoordinatorRegistryCenter zookeeperRegistryCenter) {
        this.zookeeperRegistryCenter = zookeeperRegistryCenter;
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {

        ApplicationContext applicationContext = event.getApplicationContext();
        Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(ElasticJobConfig.class);
        beanMap.forEach((key, value) -> {
            Class<?> clz = value.getClass();
            // 如果类中有内部类
            int outClzIndex = clz.getName().indexOf("$");
            if (outClzIndex > 0) {
                String clzName = clz.getName();
                try {
                    clz = Class.forName(clzName.substring(0, outClzIndex));
                } catch (ClassNotFoundException e) {
                    log.error("elastic job 启动异常, 系统强制退出", e);
                    System.exit(1);
                }
            }

            // 获取配置项 ElasticJobConfig
            ElasticJobConfig conf = clz.getAnnotation(ElasticJobConfig.class);
            // 获取配置内容
            String jobType = conf.jobType();
            String jobName = conf.jobName();
            String cron = conf.cron();
            String jobParameter = conf.jobParameter();
            String shardingItemParameters = conf.shardingItemParameters();
            String description = conf.description();
            String jobErrorHandlerType = conf.jobErrorHandlerType();
            String jobExecutorServiceHandlerType = conf.jobExecutorServiceHandlerType();
            String jobShardingStrategyType = conf.jobShardingStrategyType();

            boolean failover = conf.failover();
            boolean misfire = conf.misfire();
            boolean overwrite = conf.overwrite();
            boolean disabled = conf.disabled();
            boolean monitorExecution = conf.monitorExecution();

            int shardingTotalCount = conf.shardingTotalCount();
            int maxTimeDiffSeconds = conf.maxTimeDiffSeconds();
            int reconcileIntervalMinutes = conf.reconcileIntervalMinutes();

            // 实例化 es_job 相关的通用配置
            JobConfiguration.Builder jobCfgBuilder = JobConfiguration
                    .newBuilder(jobName, shardingTotalCount)
                    .shardingItemParameters(shardingItemParameters)
                    .description(description)
                    .failover(failover)
                    .jobParameter(jobParameter)
                    .misfire(misfire)
                    .cron(cron)
                    .overwrite(overwrite)
                    .maxTimeDiffSeconds(maxTimeDiffSeconds)
                    .reconcileIntervalMinutes(reconcileIntervalMinutes)
                    .disabled(disabled)
                    .monitorExecution(monitorExecution)
                    .jobShardingStrategyType(jobShardingStrategyType)
                    .jobErrorHandlerType(jobErrorHandlerType)
                    .jobExecutorServiceHandlerType(jobExecutorServiceHandlerType);

            // 创建一个Spring的beanDefinition
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(ScheduleJobBootstrap.class);
            // 多例
            factory.setInitMethodName("schedule")
                    .setScope(PROTOTYPE);

            // 1.添加注册中心
            factory.addConstructorArgValue(this.zookeeperRegistryCenter);

            // 2. 添加自己的真实的任务实现类, 判断是什么类型的
            if (ElasticJobType.SCRIPT.equals(jobType)) {
                // 脚本任务
                boolean streamingProcess = conf.streamingProcess();
                String scriptCommandLine = conf.scriptCommandLine();
                jobCfgBuilder
                        .setProperty(DataflowJobProperties.STREAM_PROCESS_KEY, Boolean.toString(streamingProcess))
                        .setProperty(ScriptJobProperties.SCRIPT_KEY, scriptCommandLine);
                factory.addConstructorArgValue(ElasticJobType.SCRIPT);

            } else if (ElasticJobType.HTTP.equals(jobType)) {
                // http任务
                String httpUrl = conf.httpUrl();
                String httpMethod = conf.httpMethod();
                String httpData = conf.httpData();
                String httpConnectTimeoutMilliseconds = conf.httpConnectTimeoutMilliseconds();
                String httpReadTimeoutMilliseconds = conf.httpReadTimeoutMilliseconds();
                String httpContentType = conf.httpContentType();
                jobCfgBuilder
                        .setProperty(HttpJobProperties.URI_KEY, httpUrl)
                        .setProperty(HttpJobProperties.METHOD_KEY, httpMethod)
                        .setProperty(HttpJobProperties.DATA_KEY, httpData)
                        .setProperty(HttpJobProperties.CONNECT_TIMEOUT_KEY, httpConnectTimeoutMilliseconds)
                        .setProperty(HttpJobProperties.READ_TIMEOUT_KEY, httpReadTimeoutMilliseconds)
                        .setProperty(HttpJobProperties.CONTENT_TYPE_KEY, httpContentType);
                factory.addConstructorArgValue(ElasticJobType.HTTP);
            } else {
                factory.addConstructorArgValue(value);
            }

            // 3. 添加配置
            factory.addConstructorArgValue(jobCfgBuilder.build());

            // 把 factory 也就是 ScheduleJobBootstrap 注入到 Spring 容器中
            DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
            String registerBeanName = conf.jobName() + ".ScheduleJobBootstrap";
            defaultListableBeanFactory.registerBeanDefinition(registerBeanName, factory.getBeanDefinition());

            ScheduleJobBootstrap scheduler = (ScheduleJobBootstrap) applicationContext.getBean(registerBeanName);
            scheduler.schedule();

            log.info("启动 elastic-job 作业：{}", jobName);

        });

        log.info("共启动作业数: {}", beanMap.values().size());

    }

}
